package xyz.xuw.storm.helloword.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import xyz.xuw.storm.helloword.bolt.ExclamationBolt;
import xyz.xuw.storm.helloword.spout.WordSpout;

/**
 * Created by xuw on 2016/11/9.
 */
public class ExclamationTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word", new WordSpout(), 10);
        builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");
        builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

        Config config = new Config();
        config.setDebug(true);
        config.setNumWorkers(3);

        if (args != null && args.length > 0) {
            try {
                StormSubmitter.submitTopology("storm_word_demo", config, builder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            String topologyName = "local test";
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(topologyName,config,builder.createTopology());
            Utils.sleep(10000);
            cluster.killTopology(topologyName);
            cluster.shutdown();
        }
    }
}
